package com.foreveross.taskservice.engine.worker;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.dayatang.domain.InstanceFactory;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.SchedulerException;

import com.foreveross.taskservice.application.statistics.EngineStatisticsHistoryApplication;
import com.foreveross.taskservice.application.task.statsdetail.ICrawlStatsApplication;
import com.foreveross.taskservice.application.task.statsdetail.vo.CrawlStatsVO;
import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.common.mq.destination.DestinationFactory;
import com.foreveross.taskservice.core.domain.channel.GradChannelEntity;
import com.foreveross.taskservice.infra.util.DateUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

/**
 * 任务刷新归档等worker
 * 需要做的工作
	// 1、暂停所有的worker,防止有的任务还继续影响收尾效果
	// 2、将MQ上剩下的任务回滚（包括渠道队列和回滚队列）
	// 3、cache回收与持久化
	// 4、统计当日情况报表(如果有需要)
	// 5、更新template的下次抓取时间
	// 6、继续所有的worker
 * @author Administrator
 *
 */
public class TaskRefreshWorker extends AbstractTaskStatefulJobWorker{
	
	
	@Override
	public void execute(JobExecutionContext context)
			throws JobExecutionException {
		long stime = System.currentTimeMillis();
		log.info("任务清理TaskRefreshWorker开始....");
		try{
			log.info("暂停所有worker");
			context.getScheduler().pauseAll();
			
			CacheDataSpace space=CacheDataSpace.AirFreightSpace();
			List<GradChannelEntity> channels=getGradChannelApplication().getAllChannel();
//			cleanTaskQueue(space,channels);
			cleanRollQueue(space,channels);
			//20140602 guokenye 持久化方案改变
			//采取批量取出的方式进行批量持久化，不然容易内存溢出
			//persistCache(space,channels);
			
			
			//***********************************************************************
			//执行化第一层
			List<CrawlStatsVO> vos =Lists.newArrayList();
			log.info("第一层持久化开始执行....");
			vos=InstanceFactory.getInstance(ICrawlStatsApplication.class).showAllChannelCrawlSuccessStats();
			if(vos.size()>0){
				InstanceFactory.getInstance(EngineStatisticsHistoryApplication.class).addDataByDate(vos);	
			}
			
			
			
			
			
			//删除昨天的队列名，每次只保留当天与昨天两天的
			//super.getTaskToMQApplication().clearHistory();
			
			log.info("更新所有模版启动");
			
			//将所有任务模板下一周期时间更新为第二天00:00:00
			getTaskTemplateApplication().updateAllTemplateNextGrabTime(DateUtil.addDay(DateUtil.getEarlyMorning(), 1));
			log.info("更新所有模版结束");
			
			log.info("flushCache启动");
			getTaskCacheFlush().flushDB();
			log.info("flushCache结束");
		}catch(Exception e){
			log.error(ExceptionUtils.getFullStackTrace(e));
		}finally{
			try {
				log.info("重新启动所有worker");
				context.getScheduler().resumeAll();
			} catch (SchedulerException e) {
				log.error(ExceptionUtils.getFullStackTrace(e));
			}
			log.info("任务清理TaskRefreshWorker结束....耗时ms:"+(System.currentTimeMillis()-stime));
		}
	}
	
	/**
	 * 清理任务队列
	 */
	private void cleanTaskQueue(CacheDataSpace space,List<GradChannelEntity> channels){
		try{
			log.info("清理MQ渠道队列启动");
			for(GradChannelEntity channel:channels){//回收每一个渠道的任务
				int size= (int) getMqInfoApplication().getTaskQueueSize(space, channel.getId());
				if(size>0){
					List<AbstractTaskModel> models=getRollbackTaskFromMQApplication().rolls(DestinationFactory.getInstance().getChannelDestination(space, channel.getId()), size);
					if(CollectionUtils.isNotEmpty(models)){
						log.info(String.format("渠道[%s]从MQ回收任务[%s]", channel.getName(),models.size()));
					}
				}//对于渠道的任务，丢弃
			}
			log.info("清理MQ渠道队列结束");
		}catch(Exception e){
			log.error(ExceptionUtils.getFullStackTrace(e));
		}
	}

	/**
	 * 清理回滚队列
	 */
	@SuppressWarnings("unchecked")
	private void cleanRollQueue(CacheDataSpace space,List<GradChannelEntity> channels){
		try{
			log.info("回收MQ回滚队列启动");
			int size=(int) getMqInfoApplication().getRollQueueSize(space);
			if(size>0){
				List<AbstractTaskModel> models=getRollbackTaskFromMQApplication().rolls(DestinationFactory.getInstance().getRollbackDestination(space), size);
				if(CollectionUtils.isNotEmpty(models)){
					getTaskCacheStorage().rollbacks(Sets.newLinkedHashSet(models));
				}
			}
			log.info("回收MQ回滚队列结束");
		}catch(Exception e){
			log.error(ExceptionUtils.getFullStackTrace(e));
		}
	}
	


	
	
	
	
}
